package com.rtf.framework.mq.common;

import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.rtf.framework.mq.kafka.consumer.KafkaConsumerConfiger;
import com.rtf.framework.mq.rocketmq.consumer.RocketMqConsumerConfiger;

@ConditionalOnProperty(prefix = "app.mq.consumer", value = "enable", havingValue = "true")
@EnableConfigurationProperties(MessageConsumerProperties.class)
@Configuration
public class MessageConsumerConfiger {

	/** 启动rocket消费者*/
	@ConditionalOnProperty(prefix = "app.mq.consumer.rocket", value = "enable", havingValue = "true")
	@Bean(destroyMethod = "destroy")
	public RocketMqConsumerConfiger rocketMqConsumerConfiger(MessageConsumerProperties messageConsumerProperties) {
		RocketMqConsumerConfiger rocketMqConsumerConfiger = new RocketMqConsumerConfiger(messageConsumerProperties.getRocket(), messageConsumerProperties.getCosumer());
		rocketMqConsumerConfiger.start();
		return rocketMqConsumerConfiger;
	}

	/** 启动kafka消费者*/
	@ConditionalOnProperty(prefix = "app.mq.consumer.kafka", value = "enable", havingValue = "true")
	@Bean(destroyMethod = "destroy")
	public KafkaConsumerConfiger kafkaConsumerConfiger(MessageConsumerProperties messageConsumerProperties) {
		KafkaConsumerConfiger kafkaConsumerConfiger = new KafkaConsumerConfiger(messageConsumerProperties.getKafka(), messageConsumerProperties.getCosumer());
		kafkaConsumerConfiger.start();
		return kafkaConsumerConfiger;
	}
}
